package com.atguigu.app.dwd.db;

import com.atguigu.common.EduConstant;
import com.atguigu.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author hangyangfei
 * 交易域支付成功事务事实表
 */
public class DwdTradePayDetailSuc {
    public static void main(String[] args) {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTime
        out( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
         */

        // TODO 3 读取kafka的topic_db数据
        String topicName = "topic_db";
        String groupID = "dwd_trade_pay_detail_suc";
        tableEnv.executeSql("CREATE TABLE topic_db (\n" +
                "  `database` string,\n" +
                "  `table` string,\n" +
                "  `type` string,\n" +
                "  `ts` bigint,\n" +
                "  `xid` bigint,\n" +
                "  `commit` string,\n" +
                "  `data` map<string,string>,\n" +
                "  `old` map<string,string>, \n" +
                "   pt AS PROCTIME() \n" +
                ") " + KafkaUtil.getKafkaDDL(topicName, groupID));

        // TODO 4 过滤出支付成功数据
        Table filterTable = tableEnv.sqlQuery("select\n" +
                "   `data`['id'] id,\n" +
                "   `data`['out_trade_no'] out_trade_no,\n" +
                "   `data`['order_id'] order_id,\n" +
                "   `data`['alipay_trade_no'] alipay_trade_no,\n" +
                "   `data`['total_amount'] total_amount,\n" +
                "   `data`['trade_body'] trade_body,\n" +
                "   `data`['payment_type'] payment_type,\n" +
                "   `data`['payment_status'] payment_status,\n" +
                "   `data`['create_time'] create_time,\n" +
                "   `data`['update_time'] update_time,\n" +
                "   `data`['callback_content'] callback_content,\n" +
                "   `data`['callback_time'] callback_time," +
                "   ts \n" +
                "from topic_db\n" +
                "where `table`='payment_info'\n" +
                "and `data`['payment_status']='1602'"  );
        tableEnv.createTemporaryView("payment_info",filterTable);
//        filterTable.execute().print();

        // TODO 5 读取order_detail数据
        tableEnv.executeSql("create table order_detail(\n" +
                "  id string,\n" +
                "  order_id string,\n" +
                "  course_id string,\n" +
                "  course_name string,\n" +
                "  user_id string,\n" +
                "  origin_amount string,\n" +
                "  coupon_reduce string,\n" +
                "  final_amount string,\n" +
                "  create_time string,\n" +
                "  date_id string,\n" +
                "  od_ts bigint,\n" +
                "  order_status string,\n" +
                "  province_id string,\n" +
                "  session_id string,\n" +
                "  expire_time string,\n" +
                "  expire_time_date string,\n" +
                "  oi_ts bigint,\n" +
                "  pt TIMESTAMP_LTZ(3),\n" +
                "  row_op_ts TIMESTAMP_LTZ(3)\n" +
                ")" + KafkaUtil.getKafkaDDL("dwd_trade_order_detail",groupID));
//        tableEnv.sqlQuery("select * from order_detail").execute().print();

        // TODO 7 进行join操作
        Table payTable = tableEnv.sqlQuery("select \n" +
                "   pi.id,\n" +
                "   od.id order_detail_id,\n" +
                "   od.order_id,\n" +
                "   od.course_id,\n" +
                "   od.course_name,\n" +
                "   od.user_id,\n" +
                "   od.origin_amount,\n" +
                "   od.coupon_reduce,\n" +
                "   od.final_amount,\n" +
                "   od.create_time,\n" +
                "   od.date_id,\n" +
                "   od.od_ts,\n" +
                "   od.order_status  ,\n" +
                "   od.province_id,\n" +
                "   od.session_id,\n" +
                "   od.expire_time,\n" +
                "   od.expire_time_date,\n" +
                "   od.oi_ts,\n" +
                "   od.pt,\n" +
                "   od.row_op_ts,\n" +
                "   pi.payment_type,\n" +
                "   pi.callback_time,\n" +
                "   pi.ts*1000 ts\n" +
                "from payment_info pi\n" +
                "join order_detail od\n" +
                "on od.order_id=pi.order_id\n" );

        tableEnv.createTemporaryView("pay_table",payTable);

//        tableEnv.sqlQuery("select * from pay_table").execute().print();z


        // TODO 8 写出到新的kafka主题中
        tableEnv.executeSql("create table pay_success(\n" +
                "   id string,\n" +
                "   order_detail_id string,\n" +
                "   order_id string,\n" +
                "   course_id string,\n" +
                "   course_name string,\n" +
                "   user_id string,\n" +
                "   origin_amount string,\n" +
                "   coupon_reduce string,\n" +
                "   final_amount string,\n" +
                "   create_time string,\n" +
                "   date_id string,\n" +
                "   od_ts bigint,\n" +
                "   order_status string,\n" +
                "   province_id string,\n" +
                "   session_id string,\n" +
                "   expire_time string,\n" +
                "   expire_time_date string,\n" +
                "   oi_ts bigint,\n" +
                "   pt TIMESTAMP_LTZ(3),\n" +
                "   row_op_ts TIMESTAMP_LTZ(3),\n" +
                "   payment_type string,\n" +
                "   callback_time string,\n" +
                "   ts bigint,\n" +
                "   PRIMARY KEY (id) NOT ENFORCED\n" +
                ")" + KafkaUtil.getUpsertKafkaSinkDDL("dwd_trade_pay_detail_suc"));

        tableEnv.executeSql("insert into pay_success select * from pay_table");

    }
}
